package com.atguigu.app.dwd.db;

import com.atguigu.Util.MyKafkaUtil;
import com.atguigu.Util.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author hjy
 * @create 2023/3/14 8:54
 */
//数据流:app-->mysql-->maxwell-->kafka(topic_db)-->flinkAPP-->kafka(dwd)
//程 序:mock-->mysql-->maxwell-->kafka(zk)-->Dwd06_TradeOrderRefund-->kafka(zk)
public class Dwd06_TradeOrderRefund {
    public static void main(String[] args) {
        //todo 1 创建执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //        env.enableCheckpointing(5000);
//        env.getCheckpointConfig().setCheckpointStorage("hdfs://hadoop102:8020/gmall-flink/check");
//        env.getCheckpointConfig().setCheckpointTimeout(60000);
//        env.setStateBackend(new HashMapStateBackend());
        //        System.setProperty("HADOOP_USER_NAME", "atguigu");
        //todo 2 从kafka读取数据(topic_db)
        tableEnv.executeSql(MyKafkaUtil.getTopicDBDDL("dwd_trade_order_refund"));
        //todo 3 获取退单信息表
        Table orderRefundTable = tableEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['order_id'] order_id,\n" +
                "`data`['sku_id'] sku_id,\n" +
                "`data`['refund_type'] refund_type,\n" +
                "`data`['refund_num'] refund_num,\n" +
                "`data`['refund_amount'] refund_amount,\n" +
                "`data`['refund_reason_type'] refund_reason_type,\n" +
                "`data`['refund_reason_txt'] refund_reason_txt,\n" +
                "`data`['refund_status'] refund_status,\n" +
                "`data`['create_time'] create_time,\n" +
                "pt\n" +
                "from \n" +
                "  topic_db\n" +
                "where\n" +
                "`database` = 'gmall'\n" +
                "and `table`='order_refund_info'\n" +
                "and `type`='insert'");
        tableEnv.createTemporaryView("order_refund_info",orderRefundTable);
        //todo 4 获取订单表
        Table orderInfoTable = tableEnv.sqlQuery("select\n" +
                "`data`['id'] id,\n" +
                "`data`['consignee'] consignee,\n" +
                "`data`['consignee_tel'] consignee_tel,\n" +
                "`data`['total_amount'] total_amount,\n" +
                "`data`['order_status'] order_status,\n" +
                "`data`['user_id'] user_id,\n" +
                "`data`['payment_way'] payment_way,\n" +
                "`data`['delivery_address'] delivery_address,\n" +
                "`data`['order_comment'] order_comment,\n" +
                "`data`['out_trade_no'] out_trade_no,\n" +
                "`data`['trade_body'] trade_body,\n" +
                "`data`['create_time'] create_time,\n" +
                "`data`['operate_time'] operate_time,\n" +
                "`data`['expire_time'] expire_time,\n" +
                "`data`['process_status'] process_status,\n" +
                "`data`['tracking_no'] tracking_no,\n" +
                "`data`['parent_order_id'] parent_order_id,\n" +
                "`data`['img_url'] img_url,\n" +
                "`data`['province_id'] province_id,\n" +
                "`data`['activity_reduce_amount'] activity_reduce_amount,\n" +
                "`data`['coupon_reduce_amount'] coupon_reduce_amount,\n" +
                "`data`['original_total_amount'] original_total_amount,\n" +
                "`data`['feight_fee'] feight_fee,\n" +
                "`data`['feight_fee_reduce'] feight_fee_reduce,\n" +
                "`data`['refundable_time'] refundable_time\n" +
                "from \n" +
                "  topic_db \n" +
                "where\n" +
                "`database` = 'gmall'\n" +
                "and `table`='order_info'\n" +
                "and `type`='update'\n" +
                "and `data`['order_status']='1005'\n" +
                "and `old`['order_status'] is not null");
        tableEnv.createTemporaryView("order_info",orderInfoTable);
        //todo 5 获取维表
        tableEnv.executeSql(MySqlUtil.getMysqlDDl());
        //todo 6 关联4张表--->维表关联两张，一张退货原因 一张退货类型
        Table resultTable = tableEnv.sqlQuery("select \n" +
                "  ori.id,\n" +
                "  ori.user_id,\n" +
                "  ori.order_id,\n" +
                "  ori.sku_id,\n" +
                "  ori.refund_type,\n" +
                "  dic_1.dic_name refund_name,\n" +
                "  ori.refund_num,\n" +
                "  ori.refund_amount,\n" +
                "  ori.refund_reason_type,\n" +
                "  ori.refund_reason_txt,\n" +
                "  ori.refund_status,\n" +
                "  ori.create_time,\n" +
                "  oi.consignee,\n" +
                "  oi.consignee_tel,\n" +
                "  oi.total_amount,\n" +
                "  oi.order_status,\n" +
                "  dic_2.dic_name order_status_name,\n" +
                "  oi.payment_way,\n" +
                "  oi.delivery_address,\n" +
                "  oi.order_comment,\n" +
                "  oi.out_trade_no,\n" +
                "  oi.trade_body,\n" +
                "  oi.operate_time,\n" +
                "  oi.expire_time,\n" +
                "  oi.process_status,\n" +
                "  oi.tracking_no,\n" +
                "  oi.parent_order_id,\n" +
                "  oi.province_id,\n" +
                "  oi.activity_reduce_amount,\n" +
                "  oi.coupon_reduce_amount,\n" +
                "  oi.original_total_amount,\n" +
                "  oi.feight_fee,\n" +
                "  oi.feight_fee_reduce,\n" +
                "  oi.refundable_time\n" +
                "from \n" +
                "order_refund_info ori \n" +
                "join order_info oi \n" +
                "on ori.order_id=oi.id \n" +
                "join base_dic FOR SYSTEM_TIME AS OF ori.pt AS dic_1\n" +
                "on ori.refund_type=dic_1.dic_code\n" +
                "join base_dic FOR SYSTEM_TIME AS OF ori.pt AS dic_2\n" +
                "on oi.order_status=dic_2.dic_code");
        tableEnv.createTemporaryView("result_table",resultTable);
        //todo 7 创建kafka dwd层主题
        tableEnv.executeSql("create table dwd_trade_order_refund(\n" +
                "  id STRING,\n" +
                "  user_id STRING,\n" +
                "  order_id STRING,\n" +
                "  sku_id STRING,\n" +
                "  refund_type STRING,\n" +
                "  refund_name STRING,\n" +
                "  refund_num STRING,\n" +
                "  refund_amount STRING,\n" +
                "  refund_reason_type STRING,\n" +
                "  refund_reason_txt STRING,\n" +
                "  refund_status STRING,\n" +
                "  create_time STRING,\n" +
                "  consignee STRING,\n" +
                "  consignee_tel STRING,\n" +
                "  total_amount STRING,\n" +
                "  order_status STRING,\n" +
                "  order_status_name STRING,\n" +
                "  payment_way STRING,\n" +
                "  delivery_address STRING,\n" +
                "  order_comment STRING,\n" +
                "  out_trade_no STRING,\n" +
                "  trade_body STRING,\n" +
                "  operate_time STRING,\n" +
                "  expire_time STRING,\n" +
                "  process_status STRING,\n" +
                "  tracking_no STRING,\n" +
                "  parent_order_id STRING,\n" +
                "  province_id STRING,\n" +
                "  activity_reduce_amount STRING,\n" +
                "  coupon_reduce_amount STRING,\n" +
                "  original_total_amount STRING,\n" +
                "  feight_fee STRING,\n" +
                "  feight_fee_reduce STRING,\n" +
                "  refundable_time STRING\n" +
                ")"+MyKafkaUtil.getKafkaSinkDDL("dwd_trade_order_refund"));
        //todo 8 输出数据
        tableEnv.executeSql("insert into dwd_trade_order_refund select * from result_table");
    }
}
